package org.tbstore.client;

import java.io.DataOutputStream;
import java.io.IOException;

import java.net.Socket;

import java.util.List;
import java.util.zip.DataFormatException;

public class StoreServer {
    public static final int      EXIT_SUCCESS       = 0;
    public static final int      EXIT_FAILURE       = -1;
    String                       host;
    int                          port;
    Socket                       socket;
    private MessageReceivePacket mrp;
    private DataOutputStream     ops;
    boolean                      initialized;
    long                         lastConnectionTime;
    int                          fail_count;
    boolean                      isUsed             = false;

    public StoreServer(String host, int port) {
        this.host               = host;
        this.port               = port;
        this.initialized        = false;
        this.lastConnectionTime = 0;
        this.fail_count         = 0;
    }

    public void open() throws IOException, InterruptedException {
        if (lastConnectionTime < (System.currentTimeMillis() - 300000)) {
            fail_count = 0;
        }

        if (fail_count > 3) {
            return;
        }

        fail_count++;
        lastConnectionTime = System.currentTimeMillis();

        SocketThread task = new SocketThread();

        task.setDaemon(true);
        task.start();

        task.join(2000);

        if (task.isAlive()) {
            task.interrupt();
            throw new InterruptedException();
        }

        if (task.exception != null) {
            throw task.exception;
        }

        fail_count = 0;
    }

    public void close() {
        if (socket != null) {
            try {
                ops.write((byte) 0xFF);
                ops.write((byte) 0xFF);
                ops.write((byte) 0xBF);
                ops.write((byte) 0x5A);
                ops.flush();
                socket.close();
                mrp.close();
            } catch (IOException e) {
            }
        }

        socket      = null;
        initialized = false;
    }

    /*
       public void dump(byte[] array) {
              System.out.println("\n01-02-03-04-05-06-07-08-09-10-11-12-13-14-15-16");
              String s;
              String c = "";
              for (int i = 0; i < array.length; i++) {
                  s = "0" + Integer.toHexString((int) array[i]);
                  System.out.print(s.substring(s.length() - 2) + " ");
                  if (((int) array[i] >= ' ') && ((int) array[i] <= '~')) {
                      c = c + (char) array[i];
                  } else {
                      c = c + ".";
                  }
                  if ((i + 1) == array.length) {
                      while ((i % 16) != 15) {
                          System.out.print("   ");
                          i++;
                      }
                  }
                  if ((((i + 1) % 16) == 0) || ((i + 1) >= array.length)) {
                      System.out.print(" " + c + "\n");
                      c = "";
                  }
              }
              System.out.println("\n");
       }
     */
    public int putObject(byte[] key, Object object, int ttl)
            throws CachingException {
        try {
            if (!initialized) {
                open();
            }

            if (!initialized) {
                return EXIT_FAILURE;
            }

            MessageSendPacket packet = new MessageSendPacket();

            packet.setCmd(StoreConstant.CMD_PUT);
            packet.setMax_age(ttl);
            packet.setKey(key);
            packet.setData(object);

            byte[] b = packet.getBytes();

            //dump(b);
            if (b == null) {
                return EXIT_FAILURE;
            }

            synchronized (ops) {
                ops.write(b, 0, b.length);
                ops.flush();
            }

            int version   = mrp.readCShort();
            int hash_code = mrp.readCShort();
            int len       = mrp.readCInt();

            if ((version == StoreConstant.STORE_VERSION) && (len == StoreConstant.CMD_PUT)
                        && (hash_code == packet.getHashCode())) {
                return EXIT_SUCCESS;
            }
        } catch (IOException e) {
            close();
            initialized = false;
            throw new CachingException(e);
        } catch (InterruptedException e) {
            close();
            initialized = false;
            throw new CachingException(e);
        }

        return EXIT_FAILURE;
    }

    public Object getObject(byte[] key) throws CachingException {
        Object object = null;
        byte[] b = null;

        try {
            if (!initialized) {
                open();
            }

            if (!initialized) {
                return null;
            }

            MessageSendPacket packet = new MessageSendPacket();

            packet.setCmd(StoreConstant.CMD_GET);
            packet.setKey(key);
            b = packet.getBytes();

            //dump(b);
            if (b == null) {
                return null;
            }

            synchronized (ops) {
                ops.write(b, 0, b.length);
                ops.flush();
            }

            int version   = mrp.readCShort();
            int hash_code = mrp.readCShort();
            int len       = mrp.readCInt();

            if ((version != StoreConstant.STORE_VERSION) || (len > StoreConstant.MAX_DATA_LEN)) {
                close();
                initialized = false;
                throw new CachingException("HOST:" + host + ":" + port + "VERSION:" + version + ", len:" + len);
            }

            if (hash_code != packet.getHashCode()) {
                if (len > 0) {
                    b = new byte[len];
                    mrp.read(b);
                }
            } else if (len > 0) {
                int uncompr_len = mrp.readCInt();

                if (uncompr_len > StoreConstant.MAX_DATA_LEN) {
                    b = new byte[len];
                    mrp.read(b);
                } else {
                    object = mrp.getObject(len, uncompr_len);
                }
            }
        } catch (IOException e) {
            close();
            initialized = false;
            throw new CachingException(e);
        } catch (DataFormatException e) {
            throw new CachingException(e);
        } catch (ClassNotFoundException e) {
            throw new CachingException(e);
        } catch (InterruptedException e) {
            close();
            initialized = false;
            throw new CachingException(e);
        }

        return object;
    }

    public int removeObject(byte[] key) throws CachingException {
        try {
            if (!initialized) {
                open();
            }

            if (!initialized) {
                return EXIT_FAILURE;
            }

            MessageSendPacket packet = new MessageSendPacket();

            packet.setCmd(StoreConstant.CMD_REMOVE);
            packet.setKey(key);

            byte[] b = packet.getBytes();

            //dump(b);
            if (b == null) {
                return EXIT_FAILURE;
            }

            synchronized (ops) {
                ops.write(b, 0, b.length);
                ops.flush();
            }

            int version   = mrp.readCShort();
            int hash_code = mrp.readCShort();
            int len       = mrp.readCInt();

            if ((version == StoreConstant.STORE_VERSION) && (len == StoreConstant.CMD_REMOVE)
                        && (hash_code == packet.getHashCode())) {
                return EXIT_SUCCESS;
            }
        } catch (IOException e) {
            close();
            initialized = false;
            throw new CachingException(e);
        } catch (InterruptedException e) {
            close();
            initialized = false;
            throw new CachingException(e);
        }

        return EXIT_FAILURE;
    }

    public int incCount(List list) throws CachingException {
        try {
            if (!initialized) {
                open();
            }

            if (!initialized) {
                return EXIT_FAILURE;
            }

            MessageSendPacket packet = new MessageSendPacket();

            byte[] b = packet.getIncCount(list);

            //dump(b);
            if (b == null) {
                return EXIT_FAILURE;
            }

            synchronized (ops) {
                ops.write(b, 0, b.length);
                ops.flush();
            }

            int version   = mrp.readCShort();
            int hash_code = mrp.readCShort();
            int len       = mrp.readCInt();

            if ((version == StoreConstant.STORE_VERSION) && ((len & 0xff) == StoreConstant.CMD_INCNUM)
                        && (hash_code == packet.getHashCode())) {
                return ((len & 0xffffff00) >> 8);
            }
        } catch (IOException e) {
            close();
            initialized = false;
            throw new CachingException(e);
        } catch (InterruptedException e) {
            close();
            initialized = false;
            throw new CachingException(e);
        }

        return EXIT_FAILURE;
    }

    private class SocketThread extends Thread {
        public IOException exception;

        public SocketThread() {
            exception = null;
        }

        public void run() {
            try {
                socket = new Socket(host, port);
                socket.setSoTimeout(2000);

                socket.setKeepAlive(true);
                socket.setTcpNoDelay(true);
                mrp         = new MessageReceivePacket(socket.getInputStream());
                ops         = new DataOutputStream(socket.getOutputStream());
                initialized = true;
            } catch (IOException e) {
                exception = e;
            }
        }
    }

    /**
     * @return Returns the isUsed.
     */
    public boolean isUsed() {
        return isUsed;
    }

    /**
     * @param isUsed The isUsed to set.
     */
    public void setUsed(boolean isUsed) {
        this.isUsed = isUsed;
    }
}
